package com.woniuxy.producer;

import com.woniuxy.util.RabbitMQConst;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author ：admin
 * @version ：1.0
 * @date ：Created in 2025/5/16 9:20
 * @description ：限时抢购生产者类
 * @modified_by ：
 */
@Slf4j
@Component
public class TopicProducer implements RabbitTemplate.ReturnsCallback, RabbitTemplate.ConfirmCallback {
    @Resource
    private RabbitTemplate rabbitTemplate;

    private final ConcurrentHashMap<String, AtomicInteger> retryCounts = new ConcurrentHashMap<>();

    private static final int MAX_RETRIES = 3; // 最大重试次数

    /**
     * <p>通过主题交换机发送消息。</p>
     *
     * @param msg        消息
     * @param routingKey 路由
     * @param <T>        消息数据类型
     */
    public <T> void sendTopic(T msg, String routingKey) {
        rabbitTemplate.setReturnsCallback(this);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.convertAndSend(
                RabbitMQConst.TOPIC_EXCHANGE, // 交换机
                routingKey, // 路由
                msg, // 消息
                message -> { // 消息持久化
                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    return message;
                },
                new CorrelationData(UUID.randomUUID().toString()) // 唯一标识
        );
    }

    /**
     * <p>退回回调 - 消息无法路由到队列。</p>
     *
     * @param returnedMessage 退回的消息
     */
    @Override
    public void returnedMessage(@NotNull ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        String exchange = returnedMessage.getExchange();
        String routingKey = returnedMessage.getRoutingKey();
        int replyCode = returnedMessage.getReplyCode();
        String replyText = returnedMessage.getReplyText();

        // 从消息头中获取关联数据ID
        String correlationId = message.getMessageProperties().getCorrelationId();
        if (correlationId == null) correlationId = UUID.randomUUID().toString();

        // 获取重试次数
        AtomicInteger count = retryCounts.computeIfAbsent(correlationId, k -> new AtomicInteger(0));
        int retryCount = count.incrementAndGet();

        if (retryCount <= MAX_RETRIES) {
            log.info("第" + retryCount + "次重试发送消息，replyCode=" + replyCode + "，replyText=" + replyText);

            // 创建新的CorrelationData
            CorrelationData newCorrelationData = new CorrelationData(correlationId);

            // 重新发送消息
            rabbitTemplate.send(exchange, routingKey, message, newCorrelationData);
        } else {
            log.info("消息重试次数已达上限，不再重试，correlationId=" + correlationId);
            // 将消息存入记录日志
            handleFailedMessage(message, exchange, routingKey, replyCode, replyText);
        }
    }

    /**
     * <p>确认回调 - 消息无法到达交换机。</p>
     *
     * @param correlationData 唯一标识
     * @param b               消息是否成功到达交换机
     * @param s               失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        if (b) {
            log.info("消息成功发送到交换机，ID：" + (correlationData == null ? "null" : correlationData.getId()));
        } else {
            log.info("消息发送到交换机失败，ID：" + (correlationData == null ? "null" : correlationData.getId()) +
                    "，原因：" + s);
        }
    }

    /**
     * <p>消息发送失败，重试次数到达上限，将数据写入日志。</p>
     *
     * @param message 消息体
     * @param exchange 交换机
     * @param routingKey 路由
     * @param replyCode 应答码
     * @param replyText 应答体
     */
    private void handleFailedMessage(Message message, String exchange, String routingKey, int replyCode,
                                     String replyText) {
        log.info(
                "消息发送失败，存入日志记录。\n" +
                        "message=" + message +
                        "\nexchange=" + exchange +
                        "\nroutingKey=" + routingKey +
                        "\nreplyCode=" + replyCode +
                        "\nreplyText=" + replyText
        );
    }
}
